package com.liuyu.common.framework.rocketmq.service.sentinel;

import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.liuyu.common.framework.common.constant.AppConstants;
import com.liuyu.common.framework.lang.utils.ApplicationContextUtils;
import com.liuyu.common.framework.rocketmq.fallback.RocketFallbackMessage;
import com.liuyu.common.framework.rocketmq.service.spring.RocketMqMessageProducerServiceImpl;
import com.liuyu.common.framework.rocketmq.vo.MqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * rocketmq 的限流，熔断降级的处理
 *
 * @author: liuyu
 * @date: 2023/6/14 11:10
 * @version: 1
 */
@Slf4j
@Service
public class RocketSentinelCommHandler {

    private static final String ROCKET_FAIL_MESSAGE = "消息发送失败: 主题:{},标签：{},消息:{} [错误:{},{}]";
    public static final String ROCKET_FALL_BACK_LIST = AppConstants.APP_KEY + ":rocket:fall:back:list";
    public static final String ROCKET_FALL_BACK_LIST_LOCK = AppConstants.APP_KEY + ":lock:rocket:fall:back:list:lock";
    private static final String SEND_MESSAGE = "sendMessage";
    private static final String SEND_ORDER_MESSAGE = "sendOrderMessage";

    @Autowired
    private static RedisTemplate redisTemplate;
    private static final AtomicBoolean initFlag = new AtomicBoolean(false);


    /**
     * 普通消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return void
     * @author liuyu
     * @date 2022/4/11 21:53
     */
    public static <T> void sendMessageFallback(String topic, MqMessage<T> message, BlockException e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        init();
        //添加到redis中去
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage(SEND_MESSAGE, topic, null, null, message));
        RocketMqMessageProducerServiceImpl.health.set(false);
    }

    /**
     * 注意： 必须为 static 函数
     *
     * @param e
     * @return
     */
    public static <T> void sendMessageFallback(String topic, MqMessage<T> message, Throwable e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        init();
        //添加到redis中去
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage(SEND_MESSAGE, topic, null, null, message));
        RocketMqMessageProducerServiceImpl.health.set(false);
    }


    /**
     * 顺序消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return void
     * @author liuyu
     * @date 2022/4/11 21:52
     */
    public static <T> void sendOrderMessageFallBack(String topic, MqMessage<T> message, BlockException e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        init();
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage(SEND_ORDER_MESSAGE, topic, null, null, message));
        RocketMqMessageProducerServiceImpl.health.set(false);
    }

    /**
     * 顺序消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return void
     * @author liuyu
     * @date 2022/4/11 21:52
     */
    public static <T> void sendOrderMessageFallBack(String topic, MqMessage<T> message, Throwable e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        init();
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage(SEND_ORDER_MESSAGE, topic, null, null, message));
        RocketMqMessageProducerServiceImpl.health.set(false);
    }


    /**
     * 事务消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2022/4/11 21:52
     */
    public static <T> TransactionSendResult sendMessageInTransactionFallback(String topic, MqMessage<T> message, Object object, Map<String, Object> header, BlockException e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        init();
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage("sendDelayMessageFallback", topic, null, null, message));
        TransactionSendResult sendResult = new TransactionSendResult();
        sendResult.setLocalTransactionState(LocalTransactionState.UNKNOW);
        RocketMqMessageProducerServiceImpl.health.set(false);
        return sendResult;
    }


    /**
     * 事务消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2022/4/11 21:52
     */
    public static <T> TransactionSendResult sendMessageInTransactionFallback(String topic, MqMessage<T> message, Object object, Map<String, Object> header, Throwable e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        init();
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage("sendDelayMessageFallback", topic, null, null, message));
        TransactionSendResult sendResult = new TransactionSendResult();
        sendResult.setLocalTransactionState(LocalTransactionState.UNKNOW);
        RocketMqMessageProducerServiceImpl.health.set(false);
        return sendResult;
    }





    private static void init() {
        if (redisTemplate == null) {
            if (initFlag.compareAndSet(false, true)) {
                redisTemplate = (RedisTemplate) ApplicationContextUtils.getApplicationContext().getBean("redisTemplate");
            }
        }
    }

}